/*
 * Copyright (c) 2025 Li Auto Inc. and its affiliates
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <edds/dds/builtin/lookup/LookupManager.hpp>

#include <edds/rtps/builtin/data/ParticipantProxyData.h>
#include <edds/rtps/builtin/data/WriterProxyData.h>
#include <edds/rtps/builtin/data/ReaderProxyData.h>
#include <edds/rtps/writer/StatefulWriter.h>
#include <edds/rtps/writer/RTPSWriter.h>
#include <edds/rtps/reader/StatefulReader.h>
#include <edds/rtps/reader/RTPSReader.h>
#include <history/WriterHistory.h>
#include <history/ReaderHistory.h>
#include <edds/rtps/attributes/HistoryAttributes.h>
#include <edds/rtps/attributes/WriterAttributes.h>
#include <edds/rtps/attributes/ReaderAttributes.h>
#include <rtps/participant/RTPSParticipantImpl.h>
#include <edds/rtps/builtin/data/BuiltinEndpoints.hpp>
#include <edds/rtps/common/VendorId_t.hpp>
// TODO(weihanwu) Uncomment if security is implemented.
// #include <edds/rtps/common/Guid.h>
// #include <edds/rtps/security/accesscontrol/ParticipantSecurityAttributes.h>

#include <algorithm>

//#include <deps/common/TypeSupport.hpp>

#include <elog/Log.hpp>

namespace evbs {

using namespace ertps::rtps;
using vbsutil::elog::Log;

namespace edds {
namespace dds {
namespace builtin {

const ertps::rtps::SampleIdentity INVALID_SAMPLE_IDENTITY;

LookupManager::LookupManager(BuiltinProtocols* prot)
    : participant_(nullptr),
      builtin_protocols_(prot),
      builtin_request_writer_(nullptr),
      builtin_request_reader_(nullptr),
      builtin_reply_writer_(nullptr),
      builtin_reply_reader_(nullptr),
      builtin_request_writer_history_(nullptr),
      builtin_reply_writer_history_(nullptr),
      builtin_request_reader_history_(nullptr),
      builtin_reply_reader_history_(nullptr),
      server_listener_(nullptr),
      client_listener_(nullptr),
      temp_reader_proxy_data_(
          prot->mp_participantImpl->getRTPSParticipantAttributes().allocation.locators.max_unicast_locators,
          prot->mp_participantImpl->getRTPSParticipantAttributes().allocation.locators.max_multicast_locators),
      temp_writer_proxy_data_(
          prot->mp_participantImpl->getRTPSParticipantAttributes().allocation.locators.max_unicast_locators,
          prot->mp_participantImpl->getRTPSParticipantAttributes().allocation.locators.max_multicast_locators)
/* TODO Uncomment if security is implemented
 #if HAVE_SECURITY
    , builtin_request_writer_secure_(nullptr)
    , builtin_reply_writer_secure_(nullptr)
    , builtin_request_reader_secure_(nullptr)
    , builtin_reply_reader_secure_(nullptr)
    , builtin_request_writer_secure_history_(nullptr)
    , builtin_reply_writer_secure_history_(nullptr)
    , builtin_request_reader_secure_history_(nullptr)
    , builtin_reply_reader_secure_history_(nullptr)
 #endif
 */
{}

LookupManager::~LookupManager() {
    /* TODO Uncomment if security is implemented
     #if HAVE_SECURITY
        participant_->deleteUserEndpoint(builtin_request_writer_secure_);
        participant_->deleteUserEndpoint(builtin_reply_writer_secure_);
        participant_->deleteUserEndpoint(builtin_request_reader_secure_);
        participant_->deleteUserEndpoint(builtin_reply_reader_secure_);
        delete builtin_request_writer_secure_history_;
        delete builtin_reply_writer_secure_history_;
        delete builtin_request_reader_secure_history_;
        delete builtin_reply_reader_secure_history_;
     #endif
     */
    if (lookup_timer_) {
        delete lookup_timer_;
        lookup_timer_ = nullptr;
    }
    if (nullptr != builtin_reply_reader_) {
        (void)participant_->deleteUserEndpoint(builtin_reply_reader_->getGuid());
    }
    if (nullptr != builtin_reply_writer_) {
        (void)participant_->deleteUserEndpoint(builtin_reply_writer_->getGuid());
    }
    if (nullptr != builtin_request_reader_) {
        (void)participant_->deleteUserEndpoint(builtin_request_reader_->getGuid());
    }
    if (nullptr != builtin_request_writer_) {
        (void)participant_->deleteUserEndpoint(builtin_request_writer_->getGuid());
    }
    delete builtin_request_writer_history_;
    delete builtin_reply_writer_history_;
    delete builtin_request_reader_history_;
    delete builtin_reply_reader_history_;

    delete client_listener_;
    delete server_listener_;
}

bool LookupManager::request_sync_to_server() {
    std::lock_guard<std::mutex> types_guard(types_lock_);
    if (types_to_sync_.empty()) {
        return false;
    }

    if (builtin_request_writer_->getMatchedReadersSize() == 0) {
        return false;
    }

    for (auto& type : types_to_sync_) {
        (void)get_type_info(type);
    }
    types_to_sync_.clear();
    return true;
}

bool LookupManager::type_request_period() {
    std::lock_guard<std::mutex> types_guard(types_lock_);
    if (types_to_register_.empty()) {
        return false;
    }

    if (builtin_request_writer_->getMatchedReadersSize() == 0) {
        return false;
    }

    for (auto& type : types_to_register_) {
        (void)get_type_info(type);
    }
    return true;
}

void LookupManager::restart_timer() {
    if (lookup_timer_) {
        lookup_timer_->restart_timer();
    }
}

void LookupManager::add_unreply_sample_id(const std::string& type_name, const ertps::rtps::SampleIdentity& sample_id) {
    std::lock_guard<std::mutex> lock(types_lock_);
    auto it = unReplySampleIds_.find(type_name);
    if (it == unReplySampleIds_.end()) {
        std::list<ertps::rtps::SampleIdentity> sample_id_list;
        sample_id_list.push_back(sample_id);
        unReplySampleIds_[type_name] = std::move(sample_id_list);
        return;
    }
    unReplySampleIds_[type_name].push_back(sample_id);
}

void LookupManager::type_info_reply(const std::string& type_name, const ertps::rtps::SampleIdentity& requestId) {
    Lookup_getTypeInfo_Result out;
    auto inst = TypeObjectFactory::get_instance();
    std::lock_guard<std::recursive_mutex> lock(*inst->get_type_info_mutex());
    const ertps::types::TypeInformation* info = inst->get_type_information(type_name);
    if (info == nullptr) {
        if (get_discovery_protocol() == ertps::rtps::DiscoveryProtocol::SERVER) {
            register_remote_type(type_name);
            add_unreply_sample_id(type_name, requestId);
        }
        return;
    }

    out.information = *info;
    out.type_name = type_name;

    Lookup_Reply* reply = static_cast<Lookup_Reply*>(reply_type_.create_data());
    reply->data.getTypeInfo(out);
    reply->header.requestId = requestId;

    (void)send_reply(*reply);
    reply_type_.delete_data(reply);
    logDebug(LOOKUP_SERVICE, get_builtin_request_writer_guid() << " Received type info request. " << type_name);
}

void LookupManager::remove_unreply_sample_id(const std::string& type_name) {
    std::unique_lock<std::mutex> lock(types_lock_);
    auto it = unReplySampleIds_.find(type_name);
    if (it == unReplySampleIds_.end()) {
        return;
    }
    std::list<ertps::rtps::SampleIdentity> sample_id_list = unReplySampleIds_[type_name];
    unReplySampleIds_.erase(type_name);
    lock.unlock();

    Lookup_Reply rep;
    for (auto& it : sample_id_list) {
        rep.header.requestId = it;
        type_info_reply(type_name, it);
    }
}

bool LookupManager::init_lookup_service(RTPSParticipantImpl* participant) {
    participant_ = participant;

    // 对请求未收到回应的消息，每2秒重新请求一次。
    if (get_discovery_protocol() == ertps::rtps::DiscoveryProtocol::SUPER_CLIENT) {
        lookup_timer_ = new TimedEvent(
            participant_->getEventResource(), [&]() -> bool { return type_request_period(); }, 2000);
    }

    logDebug(LOOKUP_SERVICE, "Lookup Service initializing, guid " << get_RTPS_participant()->getGuid() << " .");
    bool retVal = create_endpoints();
    /*
     #if HAVE_SECURITY
        if (retVal)
        {
            retVal = create_secure_endpoints();
        }
     #endif
     */
    return retVal;
}

bool LookupManager::assign_remote_endpoints(const ParticipantProxyData& pdata) {
    uint32_t endp = pdata.m_availableBuiltinEndpoints;
    uint32_t auxendp = endp;

    std::lock_guard<std::mutex> data_guard(temp_data_lock_);

    // 类型发现、命令行交互不在dserver-dserver间，这里屏蔽后，减少交互报文
    if (get_discovery_protocol() == ertps::rtps::DiscoveryProtocol::SERVER &&
        pdata.m_participantName.to_string().find("DS-Server") != std::string::npos) {
        return true;
    }

    temp_writer_proxy_data_.guid().guidPrefix = pdata.m_guid.guidPrefix;
    temp_writer_proxy_data_.persistence_guid().guidPrefix = pdata.m_guid.guidPrefix;
    temp_writer_proxy_data_.set_remote_locators(pdata.metatraffic_locators, participant_, true);
    temp_writer_proxy_data_.topicKind(NO_KEY);
    temp_writer_proxy_data_.m_qos.m_durability.kind = ertps::VOLATILE_DURABILITY_QOS;
    if (pdata.m_VendorId == vbsutil::xmlparser::c_VendorId_mvbs) {
        temp_writer_proxy_data_.m_qos.m_reliability.kind = ertps::BEST_EFFORT_RELIABILITY_QOS;
    } else {
        temp_writer_proxy_data_.m_qos.m_reliability.kind = ertps::RELIABLE_RELIABILITY_QOS;
    }

    temp_reader_proxy_data_.clear();
    // temp_reader_proxy_data_.m_expectsInlineQos = false;
    temp_reader_proxy_data_.guid().guidPrefix = pdata.m_guid.guidPrefix;
    temp_reader_proxy_data_.set_remote_locators(pdata.metatraffic_locators, participant_, true);
    temp_reader_proxy_data_.topicKind(NO_KEY);
    temp_reader_proxy_data_.m_qos.m_durability.kind = ertps::VOLATILE_DURABILITY_QOS;
    if (pdata.m_VendorId == vbsutil::xmlparser::c_VendorId_mvbs) {
        temp_reader_proxy_data_.m_qos.m_reliability.kind = ertps::BEST_EFFORT_RELIABILITY_QOS;
    } else {
        temp_reader_proxy_data_.m_qos.m_reliability.kind = ertps::RELIABLE_RELIABILITY_QOS;
    }

    logDebug(LOOKUP_SERVICE,
             "local participant " << get_RTPS_participant()->getGuid() << " for remote participant: " << pdata.m_guid);
    auxendp &= BUILTIN_ENDPOINT_LOOKUP_SERVICE_REQUEST_DATA_WRITER;

    if ((auxendp != 0) && (builtin_request_reader_ != nullptr)) {
        logDebug(LOOKUP_SERVICE, "Adding remote writer to the local Builtin Request Reader");
        temp_writer_proxy_data_.guid().entityId = ertps::rtps::c_EntityId_Lookup_request_writer;
        temp_writer_proxy_data_.persistence_guid().entityId = ertps::rtps::c_EntityId_Lookup_request_writer;
        (void)builtin_request_reader_->matched_writer_add(temp_writer_proxy_data_);
    }

    auxendp = endp;
    auxendp &= BUILTIN_ENDPOINT_LOOKUP_SERVICE_REPLY_DATA_WRITER;

    if ((auxendp != 0) && (builtin_reply_reader_ != nullptr)) {
        logDebug(LOOKUP_SERVICE, "Adding remote writer to the local Builtin Reply Reader");
        temp_writer_proxy_data_.guid().entityId = ertps::rtps::c_EntityId_Lookup_reply_writer;
        temp_writer_proxy_data_.persistence_guid().entityId = ertps::rtps::c_EntityId_Lookup_reply_writer;
        (void)builtin_reply_reader_->matched_writer_add(temp_writer_proxy_data_);
    }

    auxendp = endp;
    auxendp &= BUILTIN_ENDPOINT_LOOKUP_SERVICE_REQUEST_DATA_READER;

    if ((auxendp != 0) && (builtin_request_writer_ != nullptr)) {
        logDebug(LOOKUP_SERVICE, "Adding remote reader to the local Builtin Request Writer");
        temp_reader_proxy_data_.guid().entityId = ertps::rtps::c_EntityId_Lookup_request_reader;
        (void)builtin_request_writer_->matched_reader_add(temp_reader_proxy_data_);
    }

    auxendp = endp;
    auxendp &= BUILTIN_ENDPOINT_LOOKUP_SERVICE_REPLY_DATA_READER;

    if ((auxendp != 0) && (builtin_reply_writer_ != nullptr)) {
        logDebug(LOOKUP_SERVICE, "Adding remote reader to the local Builtin Reply Writer");
        temp_reader_proxy_data_.guid().entityId = ertps::rtps::c_EntityId_Lookup_reply_reader;
        (void)builtin_reply_writer_->matched_reader_add(temp_reader_proxy_data_);
    }

    request_sync_to_server();
    return true;
}

void LookupManager::remove_remote_endpoints(ertps::rtps::ParticipantProxyData* pdata) {
    GUID_t tmp_guid;
    tmp_guid.guidPrefix = pdata->m_guid.guidPrefix;

    logDebug(LOOKUP_SERVICE,
             "local participant " << get_RTPS_participant()->getGuid() << " for remote participant: " << pdata->m_guid);
    uint32_t endp = pdata->m_availableBuiltinEndpoints;
    uint32_t partdet = endp;
    uint32_t auxendp = endp;
    partdet &= DISC_BUILTIN_ENDPOINT_PARTICIPANT_DETECTOR;  // Habria que quitar esta linea que comprueba si tiene PDP.
    auxendp &= BUILTIN_ENDPOINT_LOOKUP_SERVICE_REQUEST_DATA_WRITER;

    bool is_matched = (((auxendp != 0) || (partdet != 0)) && (builtin_request_reader_ != nullptr));
    if (is_matched) {
        logDebug(LOOKUP_SERVICE, "Removing remote writer from the local Builtin Request Reader");
        tmp_guid.entityId = ertps::rtps::c_EntityId_Lookup_request_writer;
        (void)builtin_request_reader_->matched_writer_remove(tmp_guid);
    }

    auxendp = endp;
    auxendp &= BUILTIN_ENDPOINT_LOOKUP_SERVICE_REPLY_DATA_WRITER;
    is_matched = (((auxendp != 0U) || (partdet != 0U)) && (builtin_reply_reader_ != nullptr));
    if (is_matched) {
        logDebug(LOOKUP_SERVICE, "Removing remote writer from the local Builtin Reply Reader");
        tmp_guid.entityId = ertps::rtps::c_EntityId_Lookup_reply_writer;
        (void)builtin_reply_reader_->matched_writer_remove(tmp_guid);
    }

    auxendp = endp;
    auxendp &= BUILTIN_ENDPOINT_LOOKUP_SERVICE_REQUEST_DATA_READER;
    is_matched = (((auxendp != 0U) || (partdet != 0U)) && (builtin_request_writer_ != nullptr));
    if (is_matched) {
        logDebug(LOOKUP_SERVICE, "Removing remote reader from the local Builtin Request Writer");
        tmp_guid.entityId = ertps::rtps::c_EntityId_Lookup_request_reader;
        (void)builtin_request_writer_->matched_reader_remove(tmp_guid);
    }

    auxendp = endp;
    auxendp &= BUILTIN_ENDPOINT_LOOKUP_SERVICE_REPLY_DATA_READER;
    is_matched = (((auxendp != 0U) || (partdet != 0U)) && (builtin_reply_writer_ != nullptr));
    if (is_matched) {
        logDebug(LOOKUP_SERVICE, "Removing remote reader from the local Builtin Reply Writer");
        tmp_guid.entityId = ertps::rtps::c_EntityId_Lookup_reply_reader;
        (void)builtin_reply_writer_->matched_reader_remove(tmp_guid);
    }
}

bool LookupManager::create_endpoints() {
    const RTPSParticipantAttributes& pattr = participant_->getRTPSParticipantAttributes();

    // Built-in history attributes.
    HistoryAttributes hatt;
    hatt.initialReservedCaches = 0;
    hatt.maximumReservedCaches = 1000;
    hatt.memoryPolicy = DYNAMIC_REUSABLE_MEMORY_MODE;
    hatt.payloadMaxSize = TYPELOOKUP_DATA_MAX_SIZE;

    WriterAttributes watt;
    watt.endpoint.unicastLocatorList = builtin_protocols_->m_metatrafficUnicastLocatorList;
    watt.endpoint.multicastLocatorList = builtin_protocols_->m_metatrafficMulticastLocatorList;
    watt.endpoint.external_unicast_locators = builtin_protocols_->m_att.metatraffic_external_unicast_locators;
    watt.endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;
    watt.endpoint.remoteLocatorList = builtin_protocols_->m_initialPeersList;
    watt.matched_readers_allocation = pattr.allocation.participants;
    watt.endpoint.topicKind = ertps::rtps::NO_KEY;
    watt.endpoint.reliabilityKind = ertps::rtps::RELIABLE;
    watt.endpoint.durabilityKind = ertps::rtps::VOLATILE;

    // Built-in request writer
    if (use_client()) {
        client_listener_ = new LookupReplyListener(this);
        builtin_request_writer_history_ = new WriterHistory(hatt);

        RTPSWriter* req_writer;
        if (participant_->createWriter(&req_writer, watt, builtin_request_writer_history_, client_listener_,
                                       ertps::rtps::c_EntityId_Lookup_request_writer, true)) {
            builtin_request_writer_ = dynamic_cast<StatefulWriter*>(req_writer);
            if (builtin_request_writer_ == nullptr) {
                delete builtin_request_writer_history_;
                builtin_request_writer_history_ = nullptr;
                return false;
            }
            logDebug(LOOKUP_SERVICE, "Builtin lookup request writer created.");
        } else {
            elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR, "lookup request writer creation failed.");
            delete builtin_request_writer_history_;
            builtin_request_writer_history_ = nullptr;
            return false;
        }
    }

    // Built-in reply writer
    if (use_server()) {
        server_listener_ = new LookupRequestListener(this);
        builtin_reply_writer_history_ = new WriterHistory(hatt);

        RTPSWriter* rep_writer;
        if (participant_->createWriter(&rep_writer, watt, builtin_reply_writer_history_, server_listener_,
                                       ertps::rtps::c_EntityId_Lookup_reply_writer, true)) {
            builtin_reply_writer_ = dynamic_cast<StatefulWriter*>(rep_writer);
            if (builtin_reply_writer_ == nullptr) {
                delete builtin_reply_writer_history_;
                builtin_reply_writer_history_ = nullptr;
                return false;
            }
            logDebug(LOOKUP_SERVICE, "Builtin lookup reply writer created.");
        } else {
            elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR, "Lookup reply writer creation failed.");
            delete builtin_reply_writer_history_;
            builtin_reply_writer_history_ = nullptr;
            return false;
        }
    }

    ReaderAttributes ratt;
    ratt.endpoint.unicastLocatorList = builtin_protocols_->m_metatrafficUnicastLocatorList;
    ratt.endpoint.multicastLocatorList = builtin_protocols_->m_metatrafficMulticastLocatorList;
    ratt.endpoint.external_unicast_locators = builtin_protocols_->m_att.metatraffic_external_unicast_locators;
    ratt.endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;
    ratt.endpoint.remoteLocatorList = builtin_protocols_->m_initialPeersList;
    ratt.matched_writers_allocation = pattr.allocation.participants;
    // ratt.expectsInlineQos = true;
    ratt.endpoint.topicKind = ertps::rtps::NO_KEY;
    ratt.endpoint.reliabilityKind = ertps::rtps::RELIABLE;
    ratt.endpoint.durabilityKind = ertps::rtps::VOLATILE;

    // Built-in request reader
    if (use_server()) {
        builtin_request_reader_history_ = new ReaderHistory(hatt);

        RTPSReader* req_reader;
        if (participant_->createReader(&req_reader, ratt, builtin_request_reader_history_, server_listener_,
                                       ertps::rtps::c_EntityId_Lookup_request_reader, true)) {
            builtin_request_reader_ = dynamic_cast<StatefulReader*>(req_reader);
            if (builtin_request_reader_ == nullptr) {
                delete builtin_request_reader_history_;
                builtin_request_reader_history_ = nullptr;
                delete server_listener_;
                server_listener_ = nullptr;
                return false;
            }
            logDebug(LOOKUP_SERVICE, "Builtin lookup request reader created.");
        } else {
            elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR, "Lookup request reader creation failed.");
            delete builtin_request_reader_history_;
            builtin_request_reader_history_ = nullptr;
            delete server_listener_;
            server_listener_ = nullptr;
            return false;
        }
    }

    // Built-in reply reader
    if (use_client()) {
        builtin_reply_reader_history_ = new ReaderHistory(hatt);

        RTPSReader* rep_reader;
        if (participant_->createReader(&rep_reader, ratt, builtin_reply_reader_history_, client_listener_,
                                       ertps::rtps::c_EntityId_Lookup_reply_reader, true)) {
            builtin_reply_reader_ = dynamic_cast<StatefulReader*>(rep_reader);
            if (builtin_reply_reader_ == nullptr) {
                delete builtin_reply_reader_history_;
                builtin_reply_reader_history_ = nullptr;
                delete client_listener_;
                client_listener_ = nullptr;
                return false;
            }
            logDebug(LOOKUP_SERVICE, "Builtin lookup reply reader created.");
        } else {
            elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR, "Lookup reply reader creation failed.");
            delete builtin_reply_reader_history_;
            builtin_reply_reader_history_ = nullptr;
            delete client_listener_;
            client_listener_ = nullptr;
            return false;
        }
    }

    return true;
}

/* TODO Implement if security is needed.
 #if HAVE_SECURITY
   bool LookupManager::create_secure_endpoints()
   {
   }
 #endif
 */

SampleIdentity LookupManager::get_type_dependencies(const GUID_t& dst_guid,
                                                    const ertps::types::TypeIdentifierSeq& id_seq) {
    SampleIdentity id = INVALID_SAMPLE_IDENTITY;
    if (use_client()) {
        Lookup_getTypeDependencies_Req in;
        in.type_ids = id_seq;
        Lookup_Request* request = static_cast<Lookup_Request*>(request_type_.create_data());
        request->data.getTypeDependencies(in);
        request->header.dest_guid = dst_guid;
        if (send_request(*request)) {
            id = request->header.requestId;
        }
        request_type_.delete_data(request);
    }
    return id;
}

SampleIdentity LookupManager::get_types(const GUID_t& dst_guid, const ertps::types::TypeIdentifierSeq& id_seq) {
    SampleIdentity id = INVALID_SAMPLE_IDENTITY;
    if (use_client()) {
        Lookup_getTypes_Req in;
        in.type_ids = id_seq;
        Lookup_Request* request = static_cast<Lookup_Request*>(request_type_.create_data());
        request->data.getTypes(in);
        request->header.dest_guid = dst_guid;
        if (send_request(*request)) {
            id = request->header.requestId;
        }
        request_type_.delete_data(request);
    }
    return id;
}

SampleIdentity LookupManager::get_type_info(const std::string& type_name) {
    SampleIdentity id = INVALID_SAMPLE_IDENTITY;
    if (use_client()) {
        Lookup_getTypeInfo_Req in;
        in.type_name = type_name;
        Lookup_Request* request = static_cast<Lookup_Request*>(request_type_.create_data());
        request->data.getTypeInfo(in);

        if (send_request(*request)) {
            id = request->header.requestId;
        }
        request_type_.delete_data(request);
    }
    return id;
}

bool LookupManager::register_remote_type(const std::string& type_name) {
    if (use_client() == false) {
        elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR, "Lookup client not opened.");
        return false;
    }
    {
        std::lock_guard<std::mutex> types_guard(types_lock_);
        if (std::find(types_to_register_.begin(), types_to_register_.end(), type_name) != types_to_register_.end()) {
            logDebug(LOOKUP_SERVICE, "Lookup type(" << type_name << ") has registered.");
            return true;
        }
        types_to_register_.push_back(type_name);
        types_to_sync_.push_back(type_name);
    }
    request_sync_to_server();
    logInfo(LOOKUP_SERVICE, "Lookup type(" << type_name << ") register success.");
    return true;
}

StatisticWrtierReaderInfos LookupManager::lookup_writer_reader_infos(const ertps::rtps::GUID_t& guid,
                                                                     uint32_t timeout) {
    if (use_client()) {
        Statistic_getInfoByGuid_Req in;
        in.guid_ = guid;
        Lookup_Request* request = static_cast<Lookup_Request*>(request_type_.create_data());
        request->data.getGuidInfo(in, Statistic_getWRInfo_Hash);
        if (get_discovery_protocol() == ertps::rtps::DiscoveryProtocol::SERVER) {
            request->header.dest_guid = guid;  // DServer——>DClient
        }
        if (send_request(*request)) {
            std::future<std::vector<StatisticWrtierReaderInfos>> response;
            {
                std::lock_guard<std::mutex> lock(promise_lock_);
                auto& promise = wr_infos_[request->header.requestId].wr_promisers_;
                response = promise.get_future();
            }
            auto status = response.wait_for(std::chrono::milliseconds(timeout));
            std::lock_guard<std::mutex> lock(promise_lock_);
            if (status == std::future_status::ready) {
                request_type_.delete_data(request);
                std::vector<StatisticWrtierReaderInfos> mulitInfos = response.get();
                StatisticWrtierReaderInfos infos;
                for (const auto& it : mulitInfos) {
                    for (const auto& msg : it) {
                        infos.push_back(msg);
                    }
                }
                return infos;
            } else {
                elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR,
                          "Lookup writer reader fail. guid " << guid << " timeout(ms):" << timeout);
            }
            wr_infos_.erase(request->header.requestId);
        }
        request_type_.delete_data(request);
    } else {
        elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR, "use_client not opened.");
    }
    return StatisticWrtierReaderInfos();
}

StatisticMatchGuids LookupManager::lookup_remote_guids(const ertps::rtps::GUID_t& guid, uint32_t timeout) {
    if (use_client()) {
        Statistic_getInfoByGuid_Req in;
        in.guid_ = guid;
        Lookup_Request* request = static_cast<Lookup_Request*>(request_type_.create_data());
        request->data.getGuidInfo(in, Statistic_getMatchGuids_Hash);
        if (get_discovery_protocol() == ertps::rtps::DiscoveryProtocol::SERVER) {
            request->header.dest_guid = guid;  // DServer——>DClient
        }
        if (send_request(*request)) {
            std::future<StatisticMatchGuids> response;
            {
                std::lock_guard<std::mutex> lock(promise_lock_);
                auto& promise = guid_promises_[request->header.requestId];
                response = promise.get_future();
            }
            auto status = response.wait_for(std::chrono::milliseconds(timeout));
            std::lock_guard<std::mutex> lock(promise_lock_);
            if (status == std::future_status::ready) {
                request_type_.delete_data(request);
                return response.get();
            } else {
                elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR,
                          "Lookup remote guids fail. guid " << guid << " timeout(ms):" << timeout);
            }
            guid_promises_.erase(request->header.requestId);
        }
        request_type_.delete_data(request);
    } else {
        elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR, "use_client not opened.");
    }
    return StatisticMatchGuids();
}

StatisticProxyInfos LookupManager::lookup_writer_reader_proxy_infos(const ertps::rtps::GUID_t& guid, uint32_t timeout) {
    if (use_client()) {
        Statistic_getInfoByGuid_Req in;
        in.guid_ = guid;
        Lookup_Request* request = static_cast<Lookup_Request*>(request_type_.create_data());
        request->data.getGuidInfo(in, Statistic_getProxyInfo_Hash);
        if (get_discovery_protocol() == ertps::rtps::DiscoveryProtocol::SERVER) {
            request->header.dest_guid = guid;  // DServer——>DClient
        }
        if (send_request(*request)) {
            std::future<StatisticProxyInfos> response;
            {
                std::lock_guard<std::mutex> lock(promise_lock_);
                auto& promise = proxy_info_promises_[request->header.requestId];
                response = promise.get_future();
            }
            auto status = response.wait_for(std::chrono::milliseconds(timeout));
            std::lock_guard<std::mutex> lock(promise_lock_);
            if (status == std::future_status::ready) {
                request_type_.delete_data(request);
                return response.get();
            } else {
                elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR,
                          "Lookup writer/reader proxy info fail. guid " << guid << " timeout(ms):" << timeout);
            }
            proxy_info_promises_.erase(request->header.requestId);
        }
        request_type_.delete_data(request);
    } else {
        elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR, "use_client not opened.");
    }
    return StatisticProxyInfos();
}

ReturnCode_t LookupManager::lookup_statistic_send_info(const ertps::rtps::GUID_t& guid, StatisticSendInfo* info,
                                                       uint32_t timeout) {
    if (use_client()) {
        Statistic_getInfoByGuid_Req in;
        in.guid_ = guid;
        Lookup_Request* request = static_cast<Lookup_Request*>(request_type_.create_data());
        request->data.getGuidInfo(in, Statistic_getSendInfo_Hash);
        if (get_discovery_protocol() == ertps::rtps::DiscoveryProtocol::SERVER) {
            request->header.dest_guid = guid;  // DServer——>DClient
        }
        if (send_request(*request)) {
            std::future<StatisticSendInfo> response;
            {
                std::lock_guard<std::mutex> lock(promise_lock_);
                auto& promise = send_promises_[request->header.requestId];
                response = promise.get_future();
            }
            auto status = response.wait_for(std::chrono::milliseconds(timeout));
            std::lock_guard<std::mutex> lock(promise_lock_);
            if (status == std::future_status::ready) {
                *info = response.get();
                request_type_.delete_data(request);
                return ReturnCode_t::RETCODE_OK;
            } else {
                elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR,
                          "Lookup send info fail. guid " << guid << " timeout(ms):" << timeout);
            }
            send_promises_.erase(request->header.requestId);
        }
        request_type_.delete_data(request);
    } else {
        elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR, "use_client not opened.");
    }
    return ReturnCode_t::RETCODE_ERROR;
}

ReturnCode_t LookupManager::lookup_statistic_recv_info(const ertps::rtps::GUID_t& guid, StatisticRecvInfo* info,
                                                       uint32_t timeout) {
    if (use_client()) {
        Statistic_getInfoByGuid_Req in;
        in.guid_ = guid;
        Lookup_Request* request = static_cast<Lookup_Request*>(request_type_.create_data());
        request->data.getGuidInfo(in, Statistic_getRecvInfo_Hash);
        if (get_discovery_protocol() == ertps::rtps::DiscoveryProtocol::SERVER) {
            request->header.dest_guid = guid;  // DServer——>DClient
        }
        if (send_request(*request)) {
            std::future<StatisticRecvInfo> response;
            {
                std::lock_guard<std::mutex> lock(promise_lock_);
                auto& promise = recv_promises_[request->header.requestId];
                response = promise.get_future();
            }
            auto status = response.wait_for(std::chrono::milliseconds(timeout));
            std::lock_guard<std::mutex> lock(promise_lock_);
            if (status == std::future_status::ready) {
                request_type_.delete_data(request);
                *info = response.get();
                return ReturnCode_t::RETCODE_OK;
            } else {
                elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR,
                          "Lookup recv info fail. guid " << guid << " timeout(ms):" << timeout);
            }
            recv_promises_.erase(request->header.requestId);
        }
        request_type_.delete_data(request);
    } else {
        elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR, "use_client not opened.");
    }
    return ReturnCode_t::RETCODE_ERROR;
}

ReturnCode_t LookupManager::lookup_statistic_writer_reader_qos(const ertps::rtps::GUID_t& guid,
                                                               StatisticWriterReaderQos* info, uint32_t timeout) {
    if (use_client()) {
        Statistic_getInfoByGuid_Req in;
        in.guid_ = guid;
        Lookup_Request* request = static_cast<Lookup_Request*>(request_type_.create_data());
        request->data.getGuidInfo(in, Statistic_getWriterReaderQos_Hash);
        if (get_discovery_protocol() == ertps::rtps::DiscoveryProtocol::SERVER) {
            request->header.dest_guid = guid;  // DServer——>DClient
        }
        if (send_request(*request)) {
            std::future<StatisticWriterReaderQos> response;
            {
                std::lock_guard<std::mutex> lock(promise_lock_);
                auto& promise = writer_qos_promises_[request->header.requestId];
                response = promise.get_future();
            }
            auto status = response.wait_for(std::chrono::milliseconds(timeout));
            std::lock_guard<std::mutex> lock(promise_lock_);
            if (status == std::future_status::ready) {
                request_type_.delete_data(request);
                *info = response.get();
                return ReturnCode_t::RETCODE_OK;
            } else {
                elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR,
                          "Lookup writer qos fail. guid " << guid << " timeout(ms):" << timeout);
            }
            writer_qos_promises_.erase(request->header.requestId);
        }
        request_type_.delete_data(request);
    } else {
        elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR, "use_client not opened.");
    }
    return ReturnCode_t::RETCODE_ERROR;
}

Log::GuidMessageBriefsMap LookupManager::lookup_statistic_message_brief(const std::string& host_name, uint32_t pid,
                                                                        const std::string& topic_name,
                                                                        uint32_t timeout) {
    if (use_client()) {
        Lookup_remoteConfig_Req in;
        in.host_name = host_name;
        in.pid = pid;
        in.param_name = topic_name;
        Lookup_Request* request = static_cast<Lookup_Request*>(request_type_.create_data());
        request->data.remoteConfig(in, Statistic_getMsgBrief_Hash);
        if (send_request(*request)) {
            std::future<Log::GuidMessageBriefsMap> response;
            {
                std::lock_guard<std::mutex> lock(promise_lock_);
                auto& promise = msg_brief_promises_[request->header.requestId];
                response = promise.get_future();
            }
            auto status = response.wait_for(std::chrono::milliseconds(timeout));
            std::lock_guard<std::mutex> lock(promise_lock_);
            if (status == std::future_status::ready) {
                request_type_.delete_data(request);
                return response.get();
            } else {
                elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR,
                          "Lookup msg brief fail. pid " << pid << " hostname " << host_name << " topic " << topic_name
                                                        << " timeout(ms):" << timeout);
            }
            msg_brief_promises_.erase(request->header.requestId);
        }
        request_type_.delete_data(request);
    } else {
        elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_NOT_ENABLED, "Lookup client not opened.");
    }
    return Log::GuidMessageBriefsMap();
}

std::string LookupManager::lookup_statistic_dserver_info(const std::string& host_name, const std::string& params,
                                                         uint32_t timeout) {
    if (use_client()) {
        Lookup_remoteConfig_Req in;
        in.host_name = host_name;
        in.param_name = params;
        Lookup_Request* request = static_cast<Lookup_Request*>(request_type_.create_data());
        request->data.remoteConfig(in, Statistic_getDServerInfo_Hash);
        if (send_request(*request)) {
            std::future<std::string> response;
            {
                std::lock_guard<std::mutex> lock(promise_lock_);
                auto& promise = dserver_info_promises_[request->header.requestId];
                response = promise.get_future();
            }
            auto status = response.wait_for(std::chrono::milliseconds(timeout));
            std::lock_guard<std::mutex> lock(promise_lock_);
            if (status == std::future_status::ready) {
                request_type_.delete_data(request);
                return response.get();
            } else {
                elogError(
                    LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR,
                    "Get DS info fail. host name " << host_name << " params " << params << " timeout(ms):" << timeout);
            }
            dserver_info_promises_.erase(request->header.requestId);
        }
        request_type_.delete_data(request);
    } else {
        elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_NOT_ENABLED, "Lookup client not opened.");
    }
    return "";
}

ReturnCode_t LookupManager::remoteConfig(const std::string& host_name, uint32_t pid, const std::string& params,
                                         uint32_t timeout) {
    if (use_client()) {
        Lookup_remoteConfig_Req in;
        in.host_name = host_name;
        in.pid = pid;
        in.param_name = params;
        Lookup_Request* request = static_cast<Lookup_Request*>(request_type_.create_data());
        request->data.remoteConfig(in, Statistic_remoteConfig_Hash);
        if (send_request(*request)) {
            std::future<ReturnCode_t> response;
            {
                std::lock_guard<std::mutex> lock(promise_lock_);
                auto& promise = remote_config_promises_[request->header.requestId];
                response = promise.get_future();
            }
            auto status = response.wait_for(std::chrono::milliseconds(timeout));
            std::lock_guard<std::mutex> lock(promise_lock_);
            if (status == std::future_status::ready) {
                request_type_.delete_data(request);
                return response.get();
            } else {
                elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_ERROR,
                          "Remote config fail. host name " << host_name << " pid " << pid << " params " << params
                                                           << " timeout(ms):" << timeout);
            }
            remote_config_promises_.erase(request->header.requestId);
        }
        request_type_.delete_data(request);
    } else {
        elogError(LOOKUP_SERVICE, RetCode_t::RETCODE_NOT_ENABLED, "Lookup client not opened.");
    }
    return ReturnCode_t::RETCODE_ERROR;
}

ReturnCode_t LookupManager::localConfig(const std::string& params) {
    std::unordered_map<std::string, std::string> params_data;
    std::stringstream ss(params);
    std::string segment;

    while (std::getline(ss, segment, ' ')) {
        size_t delimiterPos = segment.find(':');
        if (delimiterPos != std::string::npos) {
            std::string key = segment.substr(0, delimiterPos);
            std::string value = segment.substr(delimiterPos + 1);
            if (key == "logLevel") {
                try {
                    uint32_t logLevel = std::stoul(value);
                    vbsutil::elog::Log::SetVerbosity(static_cast<uint8_t>(logLevel));
                    elogWarning(STATISTIC_LOOKUP, "Remote config logLevel:" << value);
                } catch (const std::exception& e) {
                    elogWarning(STATISTIC_LOOKUP, "Remote config logLevel:" << value << " invalid." << e.what());
                    return ReturnCode_t::RETCODE_BAD_PARAMETER;
                }
            } else if (key == "logPeriod") {
                try {
                    uint32_t logPeriod = std::stoul(value);
                    vbsutil::elog::Log::setLogPeriod(logPeriod);
                    elogWarning(STATISTIC_LOOKUP, "Remote config logPeriod:" << logPeriod);
                } catch (const std::exception& e) {
                    elogWarning(STATISTIC_LOOKUP, "Remote config logPeriod:" << value << " invalid." << e.what());
                    return ReturnCode_t::RETCODE_BAD_PARAMETER;
                }
            } else if (key == "addTopicFilter") {
                try {
                    vbsutil::elog::Log::addMsgTopicFilter(value);
                    elogWarning(STATISTIC_LOOKUP, "Remote config addMsgTrace topic:" << value);
                } catch (const std::exception& e) {
                    elogWarning(STATISTIC_LOOKUP,
                                "Remote config addMsgTrace topic:" << value << " invalid." << e.what());
                    return ReturnCode_t::RETCODE_BAD_PARAMETER;
                }
            } else if (key == "briefOutputMode") {
                uint32_t outputMode = std::stoul(value);
                try {
                    vbsutil::elog::Log::setMsgBriefOutputMode(outputMode);
                    elogWarning(STATISTIC_LOOKUP, "Remote config setMsgBriefOutputMode outputMode:" << outputMode);
                } catch (const std::exception& e) {
                    elogWarning(STATISTIC_LOOKUP, "Remote config setMsgBriefOutputMode outputMode:"
                                                      << outputMode << " invalid." << e.what());
                    return ReturnCode_t::RETCODE_BAD_PARAMETER;
                }
            } else if (key == "briefOutLimit") {
                uint32_t limit = std::stoul(value);
                try {
                    vbsutil::elog::Log::setMsgBriefLimit(limit);
                    elogWarning(STATISTIC_LOOKUP, "Remote config setMsgBriefLimit limit:" << limit);
                } catch (const std::exception& e) {
                    elogWarning(STATISTIC_LOOKUP,
                                "Remote config setMsgBriefLimit limit:" << limit << " invalid." << e.what());
                    return ReturnCode_t::RETCODE_BAD_PARAMETER;
                }
            } else if (key == "msgTraceMask") {
                uint32_t msg_trace_mask = std::stoul(value);
                try {
                    vbsutil::elog::Log::setMsgTraceMask(msg_trace_mask);
                    elogWarning(STATISTIC_LOOKUP, "Remote config setMsgTraceMask msg_trace_mask:" << msg_trace_mask);
                } catch (const std::exception& e) {
                    elogWarning(STATISTIC_LOOKUP, "Remote config setMsgTraceMask msg_trace_mask:"
                                                      << msg_trace_mask << " invalid." << e.what());
                    return ReturnCode_t::RETCODE_BAD_PARAMETER;
                }
            } else {  // 后续新增远程配置的命令行，只需这里添加分支即可。
                elogWarning(STATISTIC_LOOKUP, "Remote config:param " << key << " not support.");
                return ReturnCode_t::RETCODE_UNSUPPORTED;
            }
        } else if (segment == "clearMsgBrief") {
            try {
                vbsutil::elog::Log::clearMsgBriefInfo();
                elogWarning(STATISTIC_LOOKUP, "Remote config clearMsgBrief");
            } catch (const std::exception& e) {
                elogWarning(STATISTIC_LOOKUP, "Remote config clearMsgBrief invalid." << e.what());
                return ReturnCode_t::RETCODE_BAD_PARAMETER;
            }
        } else if (segment == "clearMsgTrace") {
            try {
                vbsutil::elog::Log::clearMsgTraceInfo();
                elogWarning(STATISTIC_LOOKUP, "Remote config clearMsgTrace");
            } catch (const std::exception& e) {
                elogWarning(STATISTIC_LOOKUP, "Remote config clearMsgTrace invalid." << e.what());
                return ReturnCode_t::RETCODE_BAD_PARAMETER;
            }
        } else {
            elogWarning(STATISTIC_LOOKUP, "Remote config:params " << params << " invalid.");
            return ReturnCode_t::RETCODE_BAD_PARAMETER;
        }
    }
    return ReturnCode_t::RETCODE_OK;
}

bool LookupManager::type_has_registered(const std::string& type_name) {
    std::lock_guard<std::mutex> types_guard(types_lock_);
    return std::find(types_to_register_.begin(), types_to_register_.end(), type_name) != types_to_register_.end();
}

bool LookupManager::add_change_with_check_full(ertps::rtps::WriterHistory* history, CacheChange_t* change) {
    if (history->isFull()) {
        (void)history->remove_min_change();
    }
    return history->add_change(change);
}

std::string LookupManager::get_instanceName() const {
    std::stringstream ss;
    ss << participant_->getGuid();
    std::string str = ss.str();
    (void)std::transform(str.begin(), str.end(), str.begin(),
                         [](unsigned char c) { return static_cast<unsigned char>(std::tolower(c)); });
    (void)str.erase(std::remove(str.begin(), str.end(), '.'), str.end());
    return "dds.builtin.TOS." + str;
}

bool LookupManager::send_request(Lookup_Request& req) {
    req.header.instanceName = get_instanceName();
    (void)req.header.requestId.writer_guid(builtin_request_writer_->getGuid());
    ++request_seq_number_;  // sn start with 1
    (void)req.header.requestId.sequence_number(request_seq_number_);

    CacheChange_t* change = builtin_request_writer_->new_change(
        [&req]() { return static_cast<uint32_t>(Lookup_Request::getCdrSerializedSize(req) + 4U); }, ALIVE);

    if (change != nullptr) {
        CDRMessage_t msg(change->serializedPayload);

        bool valid = CDRMessage::addOctet(&msg, 0U);
        change->serializedPayload.encapsulation = static_cast<uint16_t>(PL_DEFAULT_ENCAPSULATION);
        msg.msg_endian = DEFAULT_ENDIAN;
        valid &= CDRMessage::addOctet(&msg, PL_DEFAULT_ENCAPSULATION);
        valid &= CDRMessage::addUInt16(&msg, 0U);

        change->serializedPayload.pos = msg.pos;
        change->serializedPayload.length = msg.length;

        SerializedPayload_t payload;
        payload.max_size = change->serializedPayload.max_size - 4U;
        payload.data = change->serializedPayload.data + 4;
        if (valid && request_type_.serialize(&req, &payload)) {
            change->serializedPayload.length += payload.length;
            change->serializedPayload.pos += payload.pos;
            change->readerGUID = req.header.dest_guid;
            payload.data = nullptr;
            builtin_request_writer_history_->set_fragments(change);
            return add_change_with_check_full(builtin_request_writer_history_, change);
        }
        payload.data = nullptr;
        (void)builtin_request_writer_->release_change(change);
    }
    return false;
}

bool LookupManager::send_reply(Lookup_Reply& rep) {
    rep.header.instanceName = get_instanceName();

    CacheChange_t* change = builtin_reply_writer_->new_change(
        [&rep]() { return static_cast<uint32_t>(Lookup_Reply::getCdrSerializedSize(rep) + 4U); }, ALIVE);

    if (change != nullptr) {
        CDRMessage_t msg(change->serializedPayload);

        bool valid = CDRMessage::addOctet(&msg, 0U);
        change->serializedPayload.encapsulation = static_cast<uint16_t>(PL_DEFAULT_ENCAPSULATION);
        msg.msg_endian = DEFAULT_ENDIAN;
        valid &= CDRMessage::addOctet(&msg, PL_DEFAULT_ENCAPSULATION);
        valid &= CDRMessage::addUInt16(&msg, 0U);

        change->serializedPayload.pos = msg.pos;
        change->serializedPayload.length = msg.length;

        SerializedPayload_t payload;
        payload.max_size = change->serializedPayload.max_size - 4U;
        payload.data = change->serializedPayload.data + 4;
        if (valid && reply_type_.serialize(&rep, &payload)) {
            change->serializedPayload.length += payload.length;
            change->serializedPayload.pos += payload.pos;
            change->readerGUID = rep.header.requestId.writer_guid();
            payload.data = nullptr;
            builtin_reply_writer_history_->set_fragments(change);
            return add_change_with_check_full(builtin_reply_writer_history_, change);
        }
        payload.data = nullptr;
        (void)builtin_reply_writer_->release_change(change);
    }
    return false;
}

bool LookupManager::recv_request(vbs::common::CacheChange_t& change, Lookup_Request& req) const {
    CDRMessage_t msg(change.serializedPayload);

    msg.msg_endian = LITTLEEND;
    // mvbs vbs cdr版本不一致，需要兼容避免cdr异常
    change.serializedPayload.encapsulation = static_cast<uint16_t>(0x0001);
    msg.pos += 4U;  // Skip encapsulation options.

    SerializedPayload_t payload;
    payload.max_size = change.serializedPayload.max_size - 4U;
    payload.length = change.serializedPayload.length - 4U;
    payload.data = change.serializedPayload.data + 4;
    bool result = request_type_.deserialize(&payload, &req);
    payload.data = nullptr;
    return result;
}

bool LookupManager::recv_reply(vbs::common::CacheChange_t& change, Lookup_Reply& rep,
                               uint16_t& continue_recvpkt_flag) const {
    CDRMessage_t msg(change.serializedPayload);
    octet flag = 0U;
    (void)CDRMessage::readOctet(&msg, &flag);
    continue_recvpkt_flag = static_cast<uint16_t>(flag);
    msg.msg_endian = LITTLEEND;
    change.serializedPayload.encapsulation = static_cast<uint16_t>(0x0001U);
    msg.pos += 3U;  // Skip encapsulation options.

    SerializedPayload_t payload;
    payload.max_size = change.serializedPayload.max_size - 4U;
    payload.length = change.serializedPayload.length - 4U;
    payload.data = change.serializedPayload.data + 4;
    bool result = reply_type_.deserialize(&payload, &rep);
    payload.data = nullptr;
    return result;
}

const ertps::rtps::GUID_t& LookupManager::get_builtin_request_writer_guid() const {
    if (nullptr != builtin_request_writer_) {
        return builtin_request_writer_->getGuid();
    }
    return c_Guid_Unknown;
}

const ertps::rtps::GUID_t& LookupManager::get_builtin_reply_writer_guid() const {
    if (nullptr != builtin_reply_writer_) {
        return builtin_reply_writer_->getGuid();
    }
    return c_Guid_Unknown;
}

}  // namespace builtin
}  // namespace dds
}  // namespace edds
}  // namespace evbs
